package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.VideoChapterBean;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;


public class DwsVideoChapterPlayWindow {



    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)
        // TODO 3. 从 Kafka dwd_trade_order_detail 主题读取订单明细数据
        String topic = "dwd_traffic_play_log";
        String groupId = "dws_video_chapter_play_window";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> source = env.addSource(kafkaConsumer);


        // TODO 4.转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = source.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        String user_id = jsonObject.getJSONObject("common").getString("uid");
                        String s_id = jsonObject.getJSONObject("common").getString("sid");
                        String ts = jsonObject.getString("ts");
                        JSONObject videoObject = jsonObject.getJSONObject("appVideo");
                        videoObject.put("user_id",user_id);
                        videoObject.put("session_id",s_id);
                        videoObject.put("ts",ts);
                        return videoObject;
                    }
                }
        );



        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<VideoChapterBean> javaBeanStream = mappedStream.map(
                jsonObj -> {
                    String videoId = jsonObj.getString("video_id");
                    String userId = jsonObj.getString("user_id");
                    String sessionId = jsonObj.getString("session_id");
                    Long durationSec = jsonObj.getLong("play_sec");
                    Long ts = jsonObj.getLong("ts") ;
                    String stt = DateFormatUtil.toYmdHms(ts);

                    VideoChapterBean videoChapterBean = VideoChapterBean.builder()
                            .userId(userId)
                            .durationSec(durationSec)
                            .videoId(videoId)
                            .sessionId(sessionId)
                            .stt(stt)
                            .ts(ts)
                            .build();
                    return videoChapterBean;
                }
        );




        // TODO 10. 设置水位线
        SingleOutputStreamOperator<VideoChapterBean> watermarkDS = javaBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<VideoChapterBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<VideoChapterBean>() {
                                    @Override
                                    public long extractTimestamp(VideoChapterBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );


        KeyedStream<VideoChapterBean, String> keyedForAggregateStream = watermarkDS.keyBy(
                new KeySelector<VideoChapterBean, String>() {

                    @Override
                    public String getKey(VideoChapterBean javaBean) throws Exception {
                        return javaBean.getSessionId();
                    }
                }
        );


        SingleOutputStreamOperator<VideoChapterBean>  withVideoDS = AsyncDataStream.unorderedWait(
                keyedForAggregateStream,
                new DimAsyncFunction<VideoChapterBean>("dim_video_info".toUpperCase()) {

                    @Override
                    public void join(VideoChapterBean javaBean, JSONObject jsonObj) throws Exception {

                        String videoName = jsonObj.getString("video_name".toUpperCase());
                        String chapterId = jsonObj.getString("chapter_id".toUpperCase());

                        javaBean.setChapterId(chapterId);
                        javaBean.setVideoName(videoName);
                    }

                    @Override
                    public String getKey(VideoChapterBean javaBean) {
                        return javaBean.getVideoId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<VideoChapterBean>  fullInfoStream = AsyncDataStream.unorderedWait(
                withVideoDS,
                new DimAsyncFunction<VideoChapterBean>("dim_chapter_info".toUpperCase()) {

                    @Override
                    public void join(VideoChapterBean javaBean, JSONObject jsonObj) throws Exception {

                        String chapterName = jsonObj.getString("chapter_name".toUpperCase());

                        javaBean.setChapterName(chapterName);
                    }

                    @Override
                    public String getKey(VideoChapterBean javaBean) {
                        return javaBean.getChapterId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );

        SinkFunction<VideoChapterBean> jdbcSink = ClickHouseUtil.<VideoChapterBean>getJdbcSink(
                "insert into dws_video_chapter_play_window values(?,?,?,?,?,?,?,?,?)"
        );
        fullInfoStream.<VideoChapterBean>addSink(jdbcSink);
        fullInfoStream.print(">>>>>>");



        env.execute();
    }

}
